package com.sxt.disruptor.chap00;

import java.util.concurrent.Executors;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

/**
 * 	主函数
 * @author Administrator
 *	disruptor.handleEventsWith(EventHandler ... handlers)，将多个EventHandler的实现类传入方法，封装成一个EventHandlerGroup，实现多消费者消费。
 *	disruptor.handleEventsWithWorkerPool(WorkHandler ... handlers)，将多个WorkHandler的实现类传入方法，封装成一个EventHandlerGroup实现多消费者消费。
 *
 *	1. 对于某一条消息m，handleEventsWith方法返回的EventHandlerGroup，Group中的每个消费者都会对m进行消费，各个消费者之间不存在竞争。
 *	handleEventsWithWorkerPool方法返回的EventHandlerGroup，Group的消费者对于同一条消息m不重复消费；
 *	也就是，如果c0消费了消息m，则c1不再消费消息m。
 *	2. 传入的形参不同。对于独立消费的消费者，应当实现EventHandler接口。
 *	对于不重复消费的消费者，应当实现WorkHandler接口。
 */
public class DisruptorMain {

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		 int bufferSize = 1024*1024;//环形队列长度，必须是2的N次方
	        EventFactory<TranslatorData> eventFactory = new TranslatorDataEventFactory();
	        /**
	         * 	定义Disruptor，基于单生产者，阻塞策略
	         */
	        Disruptor<TranslatorData> disruptor = new Disruptor<TranslatorData>(eventFactory,bufferSize, Executors.defaultThreadFactory(),ProducerType.SINGLE,new BlockingWaitStrategy());
	        /////////////////////////////////////////////////////////////////////
	        //XXX(disruptor);//这里是调用各种不同方法的地方.
	        parallel(disruptor);
	        /////////////////////////////////////////////////////////////////////
	        RingBuffer<TranslatorData> producer = disruptor.getRingBuffer();
	        /**
	         * 	输入10
	         */
	        producer.publishEvent(new LongEventTranslator(),10L);
	        producer.publishEvent(new LongEventTranslator(),100L);
	}

	
	/**
	  *	并行计算实现,c1,c2互相不依赖
	  *	消费者之间互相不依赖，并行执行，执行开始时间是一样的
	  * p --> c11
	  *   --> c21
	  */
	 public static void parallel(Disruptor<TranslatorData> disruptor){
	     disruptor.handleEventsWith(new C11EventHandler(),new C21EventHandler());
	     disruptor.start();
	 }
	 
	 /**
	  *	串行依次执行
	  * p --> c11 --> c21
	  * @param disruptor
	  */
	 public static void serial(Disruptor<TranslatorData> disruptor){
	     disruptor.handleEventsWith(new C11EventHandler()).then(new C21EventHandler());
	     disruptor.start();
	 }
	 
	 /**
	  * 菱形方式执行
	  * <br/>
	  *   --> c11
	  * p          --> c21
	  *   --> c12
	  * @param disruptor
	  */
	 public static void diamond(Disruptor<TranslatorData> disruptor){
	     disruptor.handleEventsWith(new C11EventHandler(),new C12EventHandler()).then(new C21EventHandler());
	     disruptor.start();
	 }
}
